package subscriber;

import java.io.ByteArrayInputStream;
import java.io.InputStream;

import javax.annotation.Resource;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.NamingException;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathConstants;
import javax.xml.xpath.XPathExpression;
import javax.xml.xpath.XPathFactory;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.w3c.dom.Document;
import org.w3c.dom.NodeList;


@MessageDriven(activationConfig = {
		@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Topic"),
		@ActivationConfigProperty(propertyName = "destination", propertyValue = "jms/PublishTopic") })
public class FilterObsTwo implements MessageListener {

	private static final String ID = "MB1";

	protected String jmsServerUrl = "tcp://0.0.0.0:61616";

	private static final String REPLY_QUEUE_NAME = "jms/ReplyQueue";

	private Connection connection;
	private Session session;
	private MessageProducer messageProducer;
	
	@Resource(name = "jms/ConnectionFactory")
	private ConnectionFactory connectionFactory;

	
	@Override
	public void onMessage(Message message) {
		TextMessage receivedMessage = null;
		String messageResult = "";
		try {
			// Get JMS connection and session
			Connection connection = connectionFactory.createConnection();
			Session session = connection.createSession(true, 0);

			if (message instanceof TextMessage)
				receivedMessage = (TextMessage) message;
			
			
			// Standard of reading a XML file
			DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
			factory.setNamespaceAware(true);
			DocumentBuilder builder;
			Document doc = null;
			XPathExpression expr = null;
			builder = factory.newDocumentBuilder();
			InputStream in = new ByteArrayInputStream(receivedMessage.getText().getBytes());
			doc = builder.parse(in);

			// Create a XPathFactory
			XPathFactory xFactory = XPathFactory.newInstance();

			// Create a XPath object
			XPath xpath = xFactory.newXPath();

			// Compile the XPath expression
			expr = xpath.compile("//pos[@id='"+ID+"']");
			// Run the query and get a nodeset
			Object result = expr.evaluate(doc, XPathConstants.NODESET);
			
			// Cast the result to a DOM NodeList
			NodeList nodes = (NodeList) result;
			for (int i=0; i<nodes.getLength();i++){
				NodeList childNodes = nodes.item(i).getChildNodes();
				for (int j=0; j<childNodes.getLength();j++){
					messageResult += childNodes.item(j).getNodeName() + " : ";
					messageResult += childNodes.item(j).getTextContent() + " \n\t\t";
				}
			}
			
			initialize();
			
			TextMessage textMessage = session.createTextMessage(messageResult);
			messageProducer.send(textMessage);
			
			session.close();
			connection.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	protected void initialize() throws JMSException, NamingException {
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(jmsServerUrl);
		connection = connectionFactory.createConnection();
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		Destination publisherTopic = session.createQueue(REPLY_QUEUE_NAME);
		messageProducer = session.createProducer(publisherTopic);
		messageProducer.setTimeToLive(10000);
		// messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

		connection.start();
	}
}
